package vip.shuai7boy.trafficTemp.rtmroad;


import java.io.BufferedReader;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.InputStreamReader;
import java.text.SimpleDateFormat;
import java.util.*;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.SparkConf;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.*;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.streaming.Durations;
import org.apache.spark.streaming.api.java.*;


import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.apache.kafka.common.serialization.StringDeserializer;
import scala.Tuple2;
import vip.shuai7boy.trafficTemp.conf.ConfigurationManager;
import vip.shuai7boy.trafficTemp.constant.Constants;

/**
 * 实时统计道路拥堵情况
 * (从kafka不断获取数据进行统计)
 */

public class RoadRealTimeAnalyze {
    public static void main(String[] args) throws InterruptedException {
        // 构建Spark Streaming上下文
        SparkConf conf = new SparkConf();
        conf.setAppName("AdClickRealTimeStatSpark");

        JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));
        //设置日志级别，避免重复
        jssc.sparkContext().setLogLevel("WARN");
        //下面reduceByKeyAndWindow设置了优化，则需要设置检查点，没有优化，则不用设置此检查点
        jssc.checkpoint("./checkpoint");

        Map<String, Object> kafkaParams = new HashMap<>();
        String brokers = ConfigurationManager.getProperty(Constants.KAFKA_METADATA_BROKER_LIST);
        kafkaParams.put("bootstrap.servers", brokers);
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("group.id", "MyGroupId-Traffic");
        kafkaParams.put("auto.offset.reset", "earliest");
        kafkaParams.put("enable.auto.commit", "true");


        // 构建topic 
        String kafkaTopics = ConfigurationManager.getProperty(Constants.KAFKA_TOPICS);
        String[] kafkaTopicsSplited = kafkaTopics.split(",");

        Set<String> topics = new HashSet<String>();
        for (String kafkaTopic : kafkaTopicsSplited) {
            topics.add(kafkaTopic);
        }

        /**
         * 从kafka获取数据
         */
        JavaInputDStream<ConsumerRecord<String, String>> carRealTimeLogDStream = KafkaUtils.createDirectStream(
                jssc,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topics, kafkaParams));
        /**
         * 实时计算道路的拥堵情况
         */
        realTimeCalculateRoadState(carRealTimeLogDStream);

        jssc.start();
        jssc.awaitTermination();
        jssc.stop();
    }

    /**
     * 读取文件方法
     */
    public static List<String> readFile(String path) {
        List<String> list = new ArrayList<>();
        try {

            BufferedReader br = new BufferedReader(new InputStreamReader(new FileInputStream(path)));
            String line = br.readLine();
            while (line != null) {
                list.add(line);
                line = br.readLine();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }

        return list;
    }


    private static void controlCar(final String path, JavaPairInputDStream<String, String> carRealTimeLogDStream) {

        carRealTimeLogDStream.transform(new Function<JavaPairRDD<String, String>, JavaRDD<String>>() {

            private static final long serialVersionUID = 1L;
            
            @Override
            public JavaRDD<String> call(JavaPairRDD<String, String> rdd) throws Exception {

                SparkContext context = rdd.context();
                JavaSparkContext sc = new JavaSparkContext(context);
                List<String> blackCars = readFile(path);
                final Broadcast<List<String>> broadcast = sc.broadcast(blackCars);

                List<String> value = broadcast.value();
                for (String string : value) {
                    System.out.println("broadcast value:" + string);
                }


                JavaRDD<String> map = rdd.filter(new Function<Tuple2<String, String>, Boolean>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Boolean call(Tuple2<String, String> tuple) throws Exception {
                        List<String> list = broadcast.value();
                        //tuple._2.split("\t")[3] --- car
                        return list.contains(tuple._2.split("\t")[3]);
                    }
                }).map(new Function<Tuple2<String, String>, String>() {

                    /**
                     *定义版本号
                     */
                    private static final long serialVersionUID = 1L;

                    @Override
                    public String call(Tuple2<String, String> v1) throws Exception {
                        //卡扣号，以及action_time   直接写入到数据库中
                        //每一个log 是要缉查布控的车辆的详细信息
                        String log = v1._2;
                        //通过log这一条数据，能够截取到monitorId，action_time
                        log.split("\t");
                        return v1._2;
                    }
                });
                return map;
            }

        }).print();
    }

    /**
     * 实时统计
     *
     * @param adRealTimeLogDStream
     */
    private static void realTimeCalculateRoadState(JavaInputDStream<ConsumerRecord<String, String>> adRealTimeLogDStream) {
        JavaDStream<String> roadRealTimeLog = adRealTimeLogDStream.map(new Function<ConsumerRecord<String, String>, String>() {
            //获取日志value信息
            @Override
            public String call(ConsumerRecord<String, String> cr) throws Exception {
                String key = cr.key();
                return cr.value();
            }
        });

        /**
         * 拿到车辆的信息
         * 按照日志value进行切换，拿到卡口ID和汽车速度
         * <Monitor_id,Speed>
         */

        JavaPairDStream<String, Integer> mapToPair = roadRealTimeLog.mapToPair(new PairFunction<String, String, Integer>() {


            @Override
            public Tuple2<String, Integer> call(String log) throws Exception {
                String[] split = log.split("\t");
                return new Tuple2<>(split[1], Integer.parseInt(split[5]));
            }
        });
        //获取（卡口，（汽车速度，1））格式数据
        JavaPairDStream<String, Tuple2<Integer, Integer>> monitorId2SpeedDStream =
                mapToPair.mapValues(new Function<Integer, Tuple2<Integer, Integer>>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<Integer, Integer> call(Integer speed) throws Exception {
                        return new Tuple2<>(speed, 1);
                    }
                });

        /**
         * 用优化的方式统计速度，返回的是tuple2(monitorId,(总速度，当前卡口通过的车辆总个数))
         */
        JavaPairDStream<String, Tuple2<Integer, Integer>> resultDStream =
                monitorId2SpeedDStream.reduceByKeyAndWindow(new Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {

                    private static final long serialVersionUID = 1L;


                    @Override
                    public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> v1, Tuple2<Integer, Integer> v2) throws Exception {
                        return new Tuple2<>(v1._1 + v2._1, v1._2 + v2._2);  //将新进来的数据进行累加
                    }
                }, new Function2<Tuple2<Integer, Integer>, Tuple2<Integer, Integer>, Tuple2<Integer, Integer>>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public Tuple2<Integer, Integer> call(Tuple2<Integer, Integer> v1, Tuple2<Integer, Integer> v2) throws Exception {

                        return new Tuple2<>(v1._1 - v2._1, v2._2 - v2._2);//将过期的数据排除
                    }
                }, Durations.minutes(5), Durations.seconds(5));


        /**
         * 使用reduceByKeyAndWindow  窗口大小是1分钟，如果你的application还有其他的功能点的话，另外一个功能点不能忍受这个长的延迟。权衡一下还是使用reduceByKeyAndWindow。
         */

        resultDStream.foreachRDD(new VoidFunction<JavaPairRDD<String, Tuple2<Integer, Integer>>>() {

            private static final long serialVersionUID = 1L;

            @Override
            public void call(JavaPairRDD<String, Tuple2<Integer, Integer>> rdd) throws Exception {
                /**
                 * 改变广播变量(动态修改广播变量)
                 */
//                JavaSparkContext sc = new JavaSparkContext(rdd.context());
//                List<String> list = readFile("I:\\ControlCar.txt");
//                Broadcast<List<String>> broadcast = sc.broadcast(list);


                rdd.foreachPartition(new VoidFunction<Iterator<Tuple2<String, Tuple2<Integer, Integer>>>>() {

                    private static final long serialVersionUID = 1L;

                    @Override
                    public void call(Iterator<Tuple2<String, Tuple2<Integer, Integer>>> iterator) throws Exception {
//                        List<String> value = broadcast.value();
//                        for (String s : value) {
//                            System.out.println("广播变量 = " + s);
//                        }
                        while (iterator.hasNext()) {
                            Tuple2<String, Tuple2<Integer, Integer>> tuple = iterator.next();
                            String monitor = tuple._1;
                            int speedCount = tuple._2._1;
                            int carCount = tuple._2._2;
                            SimpleDateFormat secondFormate = new SimpleDateFormat("yyyy-MM-dd hh:mm:ss");
                            System.out.println("当前时间：" + secondFormate.format(Calendar.getInstance().getTime()) +
                                    "卡扣编号：" + monitor + "车辆总数：" + carCount + "速度总数：" + speedCount + " 平均速度：" + (speedCount / carCount));
                        }
                    }
                });
            }


        });

    }
}


